Here we will generate sample trades from the RTS 2 Annex III taxonomy. Each sample trade is then enriched with the information needed run an SI calculation.
Once the trade data is assembled the the data normally provided by the regulator is synthesised.
Lastly, the SI calculations are run
The SI calculation includes a number of tests. See Article 15 (page 39) of Brussels, 25.4.2016 C(2016) 2398 final for the three derivatives tests: a, b and c.
In [1]:
# First we need to import the libraries we'll be needing
import rts2_annex3
import pandas as pd
import random
random.seed()
In [2]:
# Get the root of the RTS 2 Annex III taxonomy
root = rts2_annex3.class_root
# Get the Asset Class we would like to generate trades for
asset_class = root.asset_class_by_name("Credit Derivatives")
# Ask the Asset class to generate some sample trade
sample_trades = asset_class.make_test_samples(number=500)
In [3]:
print("Generated {count} trades. here is one example:\n".format(count=len(sample_trades)))
print(vars(random.choice(sample_trades)))
In a real firm with real trades we would need to know the LEI (Legal Entity Identifier - ISO 17442) of the legal entity which did each trade because SI status is reported distinctly for each legal entity, identified by an LEI.
Firms may do trades within a single legal entity, perhaps to move risk from one trading desk to another. These are called intra-entity trades and must be filtered out before the SI calculation. For this example we'll say that all the trades we generated are inter-entity trades (i.e. trades between distinct legal entities), so we count them all.
In this example we'll use just one LEI, and not even a valid one, but it will suffice for the example.
In [4]:
# Typically trades invole two parties, the bank and a counterparty (the client).
# For the SI calculation we just need the bank LEI.
for sample_trade in sample_trades:
sample_trade.lei = 'Our_bank_LEI'
# Print the LEI from one of the trades (they're all the same!)
print(random.choice(sample_trades).lei)
In [5]:
# We give each sample trade a trade date in a 30 day range of dates
# and an ISO week number (c.f. https://en.wikipedia.org/wiki/ISO_week_date)
import datetime
sample_dates = []
today = datetime.date.today()
for day_number in range(-30, 0):
a_date = today + datetime.timedelta(day_number)
if a_date.weekday() < 6:
sample_dates.append(a_date)
for sample_trade in sample_trades:
selected_date = random.choice(sample_dates)
sample_trade.trade_date = selected_date
sample_trade.trade_date_week = selected_date.isocalendar()[1]
# Print the one of the modified sample trades
a_trade = random.choice(sample_trades)
print(a_trade.trade_date)
print(a_trade.trade_date_week)
A MIC (Market Identifier Code - ISO 10383) is an ID for a trading venue, for example a stock exchange. The regulator is expected to provide a list of MIC values which identify venues which are recognised for the purposes of the SI calculation. Trades which are done on vs. off recognised venues are counted differently.
In [6]:
# We define our MICs. A MIC value is always 4 charcters in length. The values used
# here are made-up nonsense, but good enough for an illustration
eea_mics = ['EEA1', 'EEA2', 'EEA3']
non_eea_mics = ['OFF1', 'OFF2', 'OFF3', 'OFF4']
all_mics = eea_mics + non_eea_mics
# Add a MIC to each sample trade
for sample_trade in sample_trades:
sample_trade.mic = random.choice(all_mics)
# Print the one of the modified sample trades
a_trade = random.choice(sample_trades)
print(a_trade.mic)
In [7]:
# Own Account is a boolean; either this is a trade which the regulator views as being
# on the bank's own account, or not. I use a random boolean with a probability.
own_account_probability = 0.25
for sample_trade in sample_trades:
sample_trade.own_account = random.random() < own_account_probability
# Print the one of the modified sample trades
a_trade = random.choice(sample_trades)
print(a_trade.own_account)
In [8]:
# Client Order is also simply a boolean. Either this is a trade which was done
# in response to a client order, or not. I use a random boolean.
client_order_probability = 0.5
for sample_trade in sample_trades:
sample_trade.client_order = random.random() < client_order_probability
# Print the one of the modified sample trades
a_trade = random.choice(sample_trades)
print(a_trade.client_order)
In [9]:
# Add a random-ish Euro Notional amount of n million EUR to each trade
notional_amounts = [x * 1000000 for x in [1, 1, 1, 2, 2, 5, 10, 25]]
for sample_trade in sample_trades:
sample_trade.eur_notional = random.choice(notional_amounts)
# Print the one of the modified sample trades
a_trade = random.choice(sample_trades)
print(a_trade.eur_notional)
In [10]:
# Now classify each trade and add the JSON classification back to the trade
for sample_trade in sample_trades:
classification = root.classification_for(subject=sample_trade)
sample_trade.rts2_classification = classification
# Print the one of the modified sample trades
a_trade = random.choice(sample_trades)
print(a_trade.rts2_classification.as_json(indent=4))
The "calculation" is really a set of filters 3 filters (a, b & c as shown below), which might identify an firm as being an SI for a particular RTS 2 subclass.
The filters all focus on the count and notional sums of trades which are OTC (not traded on EEA recognised venue), own account (traded using the banks money) and in response to a client order. Here we'll call this subset of trades SI-trades.
Tests a & b also ask if a particulat RTS 2 sub class is liquid. Whether an instrument is liquid or not is determined by the regulator, and the regulator must publish this, together with the total trade count and total notional, for each sub class.
a. If the RTS 2 Annex III sub class is liquid
b. If the RTS 2 Annex III sub class is not liquid
c. If the sum of EUR notional for SI-trades is
Here we build a tiny Python application to do the SI calculation on the generated trades.
The result of the calculation is a JSON report of the RTS 2 Annex III sub classes for which our LEI is a Systematic Internaliser.
We define 3 classes:
If you want to see the report, scroll past the code to the next text box.
In [11]:
import collections
import json
class SIReport(object):
@classmethod
def for_trades(cls, trades):
new_report = cls()
new_report.add_trades(trades)
return new_report
def __init__(self):
self.trades = []
self.sub_classes = dict()
self._number_of_weeks = None
def add_trades(self, trades):
self.trades.extend(trades)
for trade in trades:
rts2_string = trade.rts2_classification.as_json()
if not rts2_string in self.sub_classes:
self.sub_classes[rts2_string] = RTS2SubClass(self, trade)
sub_class = self.sub_classes[rts2_string]
sub_class.add_trade(trade)
self._number_of_weeks = None
@property
def number_of_weeks(self):
if self._number_of_weeks is None:
min_week = min(self.trades, key=lambda t: t.trade_date_week).trade_date_week
max_week = max(self.trades, key=lambda t: t.trade_date_week).trade_date_week
self._number_of_weeks = max_week - min_week + 1
return self._number_of_weeks
def si_sub_classes(self):
return [sub_class for sub_class in self.sub_classes.values()
if sub_class.si_status()]
def report(self):
si_sub_classes = self.si_sub_classes()
report_items = [
'This LEI is an SI for {si_count} of {all_count} '
'sub classes traded over {weeks} weeks.'.format(
si_count=len(si_sub_classes),
all_count=len(self.sub_classes),
weeks=self.number_of_weeks)]
for sub_class in si_sub_classes:
report_items.append(sub_class.report())
return json.dumps(report_items, indent=4)
class RTS2SubClass(object):
def __init__(self, si_report, sample_trade):
self.si_report = si_report
self.sample_trade = sample_trade
self.is_liquid = random.random() < 0.5
self.trades = []
self._aggregations = None
@property
def aggregations(self):
# I keep all the computed results in one object so I can drop the cache
# if a trade is added
if self._aggregations is None:
self._aggregations = Aggregations(sub_class=self)
return self._aggregations
def add_trade(self, trade):
self.trades.append(trade)
self._aggregations = None
def si_status(self):
"""
This is the SI Calculation. It is applied distinctly to each sub class.
It's quite simple once everything is aggregated.
Note that these are the rules for derivatives trades only.
"""
agg = self.aggregations
if self.is_liquid \
and agg.trade_count >= (0.025 * agg.eu_trade_count) \
and agg.avg_weekly_trades >= 1:
return "SI - (a) Liquid instrument test"
if not self.is_liquid \
and agg.avg_weekly_trades >= 1:
return "SI - (b) Non-liquid instrument test"
if agg.notional_sum >= (0.25 * agg.lei_notional_sum) \
or agg.notional_sum >= (0.01 * agg.eu_notional_sum):
return "SI - (c) Notional size test"
return None
def report(self):
report_list = ['Status: {status}.'.format(status=self.si_status())]
agg_dict = vars(self.aggregations).copy()
del agg_dict['sub_class']
del agg_dict['si_trades']
report_list.append(agg_dict)
report_list.append(self.sample_trade.rts2_classification.classification_dict())
return report_list
class Aggregations(object):
"""
Each instance of Aggregations represents the subset of the trades for
a sub class which are OTC client orders on the own account of the LEI.
The SI calculation tests are with respect to this subset of the trades.
"""
def __init__(self, sub_class):
self.sub_class = sub_class
# Build the aggregations for this sub class
self.si_trades = [
trade for trade in self.sub_class.trades
if (not trade.mic in eea_mics) # OTC
and trade.own_account # Traded on own account
and trade.client_order] # in response to a client order
self.trade_count = len(self.si_trades)
self.notional_sum = sum([abs(trade.eur_notional) for trade in self.si_trades])
self.avg_weekly_trades = self.trade_count / self.sub_class.si_report.number_of_weeks
# Now I synthesise the EU figures which should really come from the regulator
self.eu_trade_count = self.trade_count * 40 + random.choice([
self.trade_count * -1, self.trade_count])
if self.notional_sum:
self.eu_notional_sum = self.notional_sum * 100 + random.choice([
self.notional_sum * -1, self.notional_sum])
else:
self.eu_notional_sum = 1
# I keep this sub class sum here so it gets flushed if trades added
self.lei_notional_sum = sum(
[abs(trade.eur_notional) for trade in self.sub_class.trades])
Having defined the classes we should be able to just run the report.
Note that the result is different every time the report is run because the EU totals are re-synthesised each time and are random. With real data the report would be stable, and indeed this test could be changed to always produce the same results; the current implementaion is intended to show some variety.
In [12]:
# First create an instance of our report
report = SIReport.for_trades(sample_trades)
# Then get the JSON report and print it
print(report.report())
In [13]:
def eu_data_for_sub_class(sub_class):
return dict(
rts2_classification=sub_class.sample_trade.rts2_classification.as_json(),
is_liquid=sub_class.is_liquid,
eu_trade_count=sub_class.aggregations.eu_trade_count,
eu_notional_sum=sub_class.aggregations.eu_notional_sum,
)
# The set of all trades (by LEI if there is more than one)
sub_classes = pd.DataFrame\
.from_records([eu_data_for_sub_class(s) for s in list(report.sub_classes.values())])\
.set_index('rts2_classification')
In [14]:
# Put the essential information for each trade into a Pandas table.
def si_details_from_sample(sample_trade):
return dict(
lei=sample_trade.lei,
trade_date=sample_trade.trade_date,
trade_date_week=sample_trade.trade_date_week,
mic=sample_trade.mic,
own_account=sample_trade.own_account,
client_order=sample_trade.client_order,
eur_notional=sample_trade.eur_notional,
rts2_classification=sample_trade.rts2_classification.as_json(),
)
all_trades = pd.DataFrame.from_records([si_details_from_sample(s) for s in sample_trades])
In [15]:
# Get the sum of all trades by RTS 2 sub class and add it as a column to the
# This should exactly match the figure in the OO report.
lei_notional_sum_series = \
all_trades[['rts2_classification', 'eur_notional']]\
.groupby(by='rts2_classification')\
.sum()
sub_classes['lei_notional_sum'] = lei_notional_sum_series
In [16]:
# Get the trades which are OTC own account client trades
si_trades = all_trades[
~all_trades.mic.isin(eea_mics)
& all_trades.own_account
& all_trades.client_order]
In [17]:
# For the SI trades, group by RTS 2 classification geting counts and notional sums
si_agg_series = si_trades[['rts2_classification', 'eur_notional']]\
.groupby(by='rts2_classification')\
.agg(['count', 'sum'])
agg_df = pd.DataFrame(si_agg_series)
agg_df.columns = ['trade_count', 'notional_sum']
sub_classes2 = pd.merge(
sub_classes.reset_index(),
agg_df.reset_index(),
how='inner', on='rts2_classification')
In [18]:
# Add a column for the average number of trades per week
min_week_number = all_trades['trade_date_week'].min()
max_week_number = all_trades['trade_date_week'].max()
number_of_weeks = max_week_number - min_week_number + 1
sub_classes3 = sub_classes2.copy()
sub_classes3['avg_weekly_trades'] = sub_classes3['trade_count']\
.apply(lambda x: x / number_of_weeks)
In [19]:
# Filter a
fa = sub_classes3.copy()
fa[(fa.is_liquid)
& (fa.trade_count >= (fa.eu_trade_count * 0.025))
& (fa.avg_weekly_trades >= 1)]
Out[19]:
In [20]:
# Filter b
fb = sub_classes3.copy()
fb[(~fb.is_liquid)
& (fb.avg_weekly_trades >= 1)]
Out[20]:
In [21]:
# Filter c
fc = sub_classes3.copy()
fc[ (fc.notional_sum >= (fc.lei_notional_sum * 0.25))
| (fc.notional_sum >= (fc.eu_notional_sum * 0.01))]
Out[21]: